package org.wejob.core.log;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.*;

public class Mqueue {
    ConcurrentHashMap<String,Topic> topics;
    ConcurrentList<Topic> topicList;
    ConcurrentList<Writer>waiters;
    int writers=0;
    //dispatcher相当于交换器exchange
    Dispatcher dispatcher;
    public Mqueue(){
        this.topics=new ConcurrentHashMap<>();
        this.dispatcher=new Dispatcher();
        this.topicList=new ConcurrentList<>();
        this.waiters=new ConcurrentList<>();
    }

    public Topic getTopic(String topic){
        return this.topics.get(topic);
    }



    public void createTipic(String topicName) throws IOException {
        if(this.topics.containsKey(topicName))
            throw new RuntimeException("topic is existed");
        doCreateTopic(topicName);
    }
    //同步方法,只能一个线程创建topic
    private synchronized Topic doCreateTopic(String topic) throws IOException {

        Topic t=this.topics.getOrDefault(topic,null);
        if(t==null){
            t=new Topic(topic);
            this.topics.put(topic,t);
            this.topicList.put(t);
            //this.dispatcher.dispatch(t);
        }
        if(this.writers<4){
            Writer writer=new Writer(this.topicList,this);
            writer.start();
            this.writers++;
            System.out.println("new writer--"+writer.getName()+" state --"+writer.getState());
        }
        return t;
    }

    public boolean put(String topic,Object message) throws IOException {
        Topic var1=checkTopic(topic);
        MessageRecord messageRecord=new MessageRecord<>(topic,null,message,System.currentTimeMillis());
        messageRecord.type=1;
        var1.messageQueue.offer(messageRecord);
        return true;
    }
    public void put(String topic, Object message,Callback callback) throws IOException {
        Topic var2=checkTopic(topic);
        MessageRecord messageRecord=new MessageRecord<>(topic,null,message,System.currentTimeMillis());
        messageRecord.callback=callback;
        messageRecord.type=2;
        var2.messageQueue.offer(messageRecord);
    }
    public void read() throws IOException, InterruptedException {
        Thread.sleep(2000);
        Topic topic=checkTopic("asynctopic3");
        IndexEntry res=topic.indexOffset.lookup(40497);
        System.out.println("lookup result  "+res.toString());
        /*ByteBuffer buffer=ByteBuffer.allocate(50);
        topic.fileChannel.read(buffer,0);

        buffer.flip();

        String decode = StandardCharsets.UTF_8.decode(buffer).toString();
        System.out.println(decode);*/
    }
    public Future asyncPut(String topic,Object message) throws InterruptedException, IOException {
        Topic var3=checkTopic(topic);
        MessageRecord messageRecord=new MessageRecord<>(topic,null,message,System.currentTimeMillis());
        messageRecord.type=3;
        messageRecord.t=var3;
        var3.messageQueue.offer(messageRecord);
        ProduceFuture future=new ProduceFuture(messageRecord,var3);
        messageRecord.future=future;
        return future;
    }

    private  Topic checkTopic(String topic) throws IOException {
        Topic var1=this.topics.getOrDefault(topic,null);
        if(var1==null){
            var1=doCreateTopic(topic);
        }
        //先判空,减少竞争
        if(!this.waiters.isEmpty()){
            Writer writer=this.waiters.poll();
            if(writer!=null){
                synchronized (writer){
                    if(!writer.getState().equals(Thread.State.RUNNABLE)){
                        System.out.println("writer "+writer.getName()+" notify");
                        writer.notify();
                    }
                }
            }
        }

        return var1;
    }


}
